package com.yytek.pm.tddb.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson2.JSON;
import com.yytek.common.client.base.MqMessage;
import com.yytek.common.client.net.IotMqttClient;
import com.yytek.common.core.constant.CacheConstants;
import com.yytek.common.core.constant.MqttConstants;
import com.yytek.common.core.domain.DeviceCache;
import com.yytek.common.core.utils.StringUtils;
import com.yytek.common.core.utils.bean.BeanCopyUtils;
import com.yytek.common.redis.service.RedisService;
import com.yytek.common.web.domain.CollectInfo;
import com.yytek.common.web.domain.ProCommonMessage;
import com.yytek.common.web.domain.ThingModelMessage;
import com.yytek.nmp.api.RemoteNmpService;
import com.yytek.tddb.api.RemoteTdEngineService;
import com.yytek.tddb.api.domain.Fields;
import com.yytek.tddb.api.domain.TableDto;
import com.yytek.tddb.api.enums.DataTypeEnum;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 处理时序数据库数据业务
 */
@Service
public class IotTddbHandleServiceImpl {
    /**
     * 最大处理队列数
     */
    private final int maxBatchSize = 100;
    @Resource
    private RemoteTdEngineService tdEngineService;
    @Resource
    private RedisService redisService;
    @Resource
    private RemoteNmpService remoteNmpService;
    @Resource
    public IotMqttClient iotMqttClient;
    /**
     * 解析并缓存数据
     * 如果达到最大队列数,触发写入数据库
     * @param line
     */
    public void processTddbLine(MqMessage line) {
        //解析数据并放入属性数据缓存队列
        paraseTddbMsgJoinQueue(line);
        //如果插入的数据队列数量大于缓存数量,批量更新
        if(getTddbLineSize() >= maxBatchSize){
            flushQueue();
        }
    }

    /**
     * 解析数据并放入属性数据缓存队列,放入告警规则mqtt队列
     * @param line
     */
    public void paraseTddbMsgJoinQueue(MqMessage line) {
        String msg = new String(line.getPayload());
        //解析line中换成接收的参数实体
        ThingModelMessage thingModelMessage= JSON.parseObject
                (msg,ThingModelMessage.class);
        //解析实体中传的参数信息放进插入数据中
        paraseMsgToTableDtoListAndSetRedis(thingModelMessage);
        //解析实体中传的参数到告警mqtt规则中
        paraseMsgToAmMqtt(thingModelMessage);
    }

    /**
     * 解析实体中传的参数分设备到告警mqtt规则
     * @param thingModelMessage
     */
    public void paraseMsgToAmMqtt(ThingModelMessage thingModelMessage) {
        if(CollectionUtil.isNotEmpty(thingModelMessage.getData())){
            List<ProCommonMessage> deviceData = thingModelMessage.getData();
            for (int i = 0; i < deviceData.size(); i++) {
                ProCommonMessage sendData = deviceData.get(i);
                //如果没有产品key,查询产品key放入
                if(StringUtils.isEmpty(sendData.getProductKey())){
                    DeviceCache deviceCache = BeanCopyUtils.mapToBean((Map<String, Object>)
                            remoteNmpService.getDeviceCache(sendData.getDeviceId()).getData(), DeviceCache.class);
                    sendData.setProductKey(deviceCache.getProductKey());
                }
                //发送到规则引擎mqtt
                iotMqttClient.publish(MqttConstants.AM_TOPIC,JSON.toJSONString(sendData));
            }
        }
    }

    /**
     * 解析参数实体中各个设备的保存字段和值
     * @param thingModelMessage
     * @return
     */
    private void paraseMsgToTableDtoListAndSetRedis(ThingModelMessage thingModelMessage) {
        if(CollectionUtil.isNotEmpty(thingModelMessage.getData())){
            List<ProCommonMessage> deviceData = thingModelMessage.getData();
            for (int i = 0; i < deviceData.size(); i++) {
                ProCommonMessage tempData = deviceData.get(i);
                TableDto tableDto = new TableDto();
                String productKey = "";
                //如果没有产品key,查询产品key放入
                if(StringUtils.isEmpty(tempData.getProductKey())){
                    DeviceCache deviceCache = BeanCopyUtils.mapToBean((Map<String, Object>)
                            remoteNmpService.getDeviceCache(tempData.getDeviceId()).getData(), DeviceCache.class);
                    productKey =deviceCache.getProductKey();
                }else{
                    productKey = tempData.getProductKey();
                }
                //设置设备表名,产品id_设备id
                tableDto.setTableName(productKey+"_"+tempData.getDeviceId());
                //取出字段值
                Map<String, CollectInfo> params = tempData.getParams();
                //不同时间段,处理不同字段队列数据
                Map<Long, ProCommonMessage> dcTimeParams = new HashMap<>();
                //循环取出相同时间段字段
                for(Map.Entry entry : params.entrySet()){
                    //取出单个上报字段值
                    CollectInfo collectInfo = params.get(entry.getKey().toString());
                    //单个属性发生时间
                    Long dcTime =collectInfo.getTime();
                    //如果单个属性发生时间存在
                    if(dcTimeParams.containsKey(dcTime)){
                        //取出字段值map
                        ProCommonMessage tempPro = dcTimeParams.get(dcTime);
                        Map<String, CollectInfo> tempParams = tempPro.getParams();
                        //添加单个字段值到map的字段信息
                        tempParams.put(entry.getKey().toString(),collectInfo);
                        //添加到字段值
                        tempPro.setParams(tempParams);
                    }else{
                        //不存在该相同时间
                        ProCommonMessage tempPro = new ProCommonMessage();
                        //添加发生时间
                        tempPro.setDctime(dcTime);
                        //添加字段值
                        Map<String, CollectInfo> tempParams = new HashMap<>();
                        //添加单个字段值到map的字段信息
                        tempParams.put(entry.getKey().toString(),collectInfo);
                        //添加到字段值
                        tempPro.setParams(tempParams);
                        //添加到不同时间段map
                        dcTimeParams.put(dcTime,tempPro);
                    }
                }
                //循环相同时间段数据,添加到字段值
                for(Map.Entry entry : dcTimeParams.entrySet()){
                    //相同时间key
                    Long key = (Long) entry.getKey();
                    //取出添加的值
                    ProCommonMessage tempPro = dcTimeParams.get(key);
                    /**
                     * 超级表普通列字段的值
                     * 值需要与创建超级表时普通列字段的数据类型对应上
                     */
                    List<Fields> schemaFieldValues = paraseMsgSchemaFields(tempPro);
                    /**
                     * 超级表标签字段的值
                     * 值需要与创建超级表时标签字段的数据类型对应上
                     */
                    List<Fields> tagsFieldValues = paraseMsgTagsFields(tempPro);
                    if(CollectionUtil.isNotEmpty(schemaFieldValues)){
                        //字段信息对象集合通过stream流过滤掉没有字段值的字段对象
                        List<Fields> schemaFieldsStream = schemaFieldValues.stream().filter(fields -> fields.getFieldValue() != null).collect(Collectors.toList());
                        tableDto.setSchemaFieldValues(schemaFieldsStream);
                    }

                    if(CollectionUtil.isNotEmpty(tagsFieldValues)){
                        //字段信息对象集合通过stream流过滤掉没有字段值的字段对象
                        List<Fields> tagsFieldsStream = tagsFieldValues.stream().filter(fields -> fields.getFieldValue() != null).collect(Collectors.toList());
                        tableDto.setTagsFieldValues(tagsFieldsStream);
                    }

                    //放入属性数据缓存队列
                    putTdInsertDataQueue(tableDto);
                }

            }

        }
    }

    /**
     * 解析标签数据放入
     * @param tempData
     * @return
     */
    private List<Fields> paraseMsgTagsFields(ProCommonMessage tempData) {
        List<Fields> o = null;
        return o;
    }

    /**
     * 解析属性数据放入
     * @param tempData
     * @return
     */
    private List<Fields> paraseMsgSchemaFields(ProCommonMessage tempData) {
        Map<String, CollectInfo> params = tempData.getParams();
        List<Fields> fields = new ArrayList<>();
        //取出事件发生时间，并格式化为long类型的毫秒时间戳
        Long eventDataTime = tempData.getDctime();
        if(eventDataTime == null){
            eventDataTime = DateUtil.current();
        }
        //超级表第一个字段数据类型必须为时间戳,默认Ts为当前系统时间
        Fields tsColumn = new Fields();
        tsColumn.setFieldName("ts");
        //以服务器时间为主
        tsColumn.setFieldValue(DateUtil.current());

        tsColumn.setDataType(DataTypeEnum.TIMESTAMP);
        fields.add(tsColumn);

        //超级表第二个字段为事件发生时间数据类型必须为时间戳
        Fields eventTimeColumn = new Fields();
        eventTimeColumn.setFieldName("event_time");
        eventTimeColumn.setFieldValue(eventDataTime);
        eventTimeColumn.setDataType(DataTypeEnum.TIMESTAMP);
        fields.add(eventTimeColumn);

        for(Map.Entry entry : params.entrySet()){
            Fields field = new Fields();
            field.setFieldName(entry.getKey().toString());
            CollectInfo collectInfo = params.get(entry.getKey().toString());
            field.setFieldValue(collectInfo.getValue());
            fields.add(field);
        }

        return fields;
    }

    /**
     * 单个设备放入属性数据缓存队列
     * @param result
     */
    public void putTdInsertDataQueue(TableDto result) {
        redisService.setCacheQueue(CacheConstants.TDDB_INSERT_QUEUE_DATA_KEY,result);
    }


    /**
     * 查询数据队列,批量插入时序数据库
     */
    public void flushQueue() {
        List<TableDto> insertList = redisService.getCacheList(CacheConstants.TDDB_INSERT_QUEUE_DATA_KEY);
        tdEngineService.insertListData(insertList);
        redisService.deleteObject(CacheConstants.TDDB_INSERT_QUEUE_DATA_KEY);
    }


    /**
     * 获取队列的长度
     * @return
     */
    public Long getTddbLineSize() {
        return redisService.getQueueSize(CacheConstants.TDDB_INSERT_QUEUE_DATA_KEY);
    }

    /**
     * 获取队列中是否有值
     * @return
     */
    public boolean hasTdQueueRedisValues() {
        return getTddbLineSize() > 0;
    }
}
